package com.nageoffer.shortlink.project.mq.consumer;

import com.nageoffer.shortlink.project.dto.biz.ShortLinkStatsRecordDTO;
import com.nageoffer.shortlink.project.service.ShortLinkService;
import lombok.RequiredArgsConstructor;
import org.redisson.api.RBlockingDeque;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;

import java.util.concurrent.Executors;
import java.util.concurrent.locks.LockSupport;

import static com.nageoffer.shortlink.project.common.constant.RedisKeyConstant.DELAY_QUEUE_STATS_KEY;

@Component
@RequiredArgsConstructor
public class DelayShortLinkStatsConsumer implements InitializingBean {

    // Redisson客户端，用于操作Redis
    private final RedissonClient redissonClient;
    // 短链接服务，用于处理统计逻辑
    private final ShortLinkService shortLinkService;

    // 消息监听方法
    public void onMessage() {
        // 创建单线程执行器（守护线程）
        Executors.newSingleThreadExecutor(
                        runnable -> {
                            Thread thread = new Thread(runnable);
                            thread.setName("delay_short-link_stats_consumer");
                            thread.setDaemon(Boolean.TRUE);
                            return thread;
                        })
                .execute(() -> {
                    // 获取Redis阻塞队列
                    RBlockingDeque<ShortLinkStatsRecordDTO> blockingDeque = redissonClient.getBlockingDeque(DELAY_QUEUE_STATS_KEY);
                    // 获取延迟队列
                    RDelayedQueue<ShortLinkStatsRecordDTO> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
                    // 无限循环处理消息
                    for (; ; ) {
                        try {
                            // 从延迟队列中获取统计记录
                            ShortLinkStatsRecordDTO statsRecord = delayedQueue.poll();
                            if (statsRecord != null) {
                                // 处理统计记录
                                shortLinkService.shortLinkStats(null, null, statsRecord);
                                continue;
                            }
                            // 队列为空时暂停500毫秒（避免CPU空转）
                            LockSupport.parkUntil(500);
                        } catch (Throwable ignored) {
                            // 捕获所有异常防止线程终止
                        }
                    }
                });
    }

    // Spring Bean初始化完成后自动调用
    @Override
    public void afterPropertiesSet() throws Exception {
        // 启动消息监听
        onMessage();
    }
}